package com.elinshaw.hadoopdev;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * @author Administrator
 */
public class HadoopDevApplication {

    /**
     * 将程序打成jar包，提交到hadoop集群中使用命令行提交任务
     * <p>
     * 对于输入输出路径，均可接受本地路径和hdfs路径
     * <p>
     * 本地路径前缀：file://
     * <p>
     * hdfs路径前缀：hdfs://
     * <p>
     * <p>
     * docker cp  /root/hadoop-dev.jar  eaa4ad75f7ea:/root/hadoop-dev.jar
     * hadoop jar hadoop-dev.jar com.elinshaw.hadoopdev.HadoopDevApplication file:/root/english.txt file:/root/output
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        Job mrJob = Job.getInstance(conf);
        //打包到集群当中时候必须指定程序的主函数
        mrJob.setJarByClass(HadoopDevApplication.class);
        //本mrJob使用的mapper和reducer的类
        mrJob.setMapperClass(TestMapper.class);
        mrJob.setReducerClass(TestReducer.class);
        //指定mapper的输出数据kv类型
        mrJob.setMapOutputKeyClass(Text.class);
        mrJob.setMapOutputValueClass(LongWritable.class);
        //指定reducer的输出数据kv类型
        //如果mapper不设置setOutputKeyClass会对mapper和reducer都起作用
        mrJob.setOutputKeyClass(Text.class);
        mrJob.setOutputValueClass(LongWritable.class);
        //设置分区类需要额外编写分区实现类
        mrJob.setPartitionerClass(MyPartitioner.class);
        mrJob.setNumReduceTasks(1);
        //设置预合并类
        mrJob.setCombinerClass(TestReducer.class);
        mrJob.setInputFormatClass(TextInputFormat.class);
        mrJob.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.addInputPath(mrJob, new Path(otherArgs[0]));
        //MapReduce输出路径结果都是输入到文件中
        FileOutputFormat.setOutputPath(mrJob, new Path(otherArgs[1]));
        boolean waitForCompletion = mrJob.waitForCompletion(true);
        //将job提交给集群运行
        System.out.println(waitForCompletion);


    }

}
